<?php
/**
 * swoole_process. 消息队列式
 * User: pushaowei
 * Date: 2016/9/11 0011
 * Time: 14:59
 */
class BaseProcess{
    private $process;
    private $process_list = []; //对应的子进程的数组
    private $process_use = []; //标记进程是否使用当中
    private $min_worker_num = 3; //最小的worker
    private $max_worker_num = 6; //最大的worker
    private $current_num; //当前worker 标记
    /**
     * BaseProcess constructor.
     */
    public function __construct(){
        $this->runMain();
    }

    public function runMain(){
        $this->process = new swoole_process([$this,'run'],false,2);//启动的一个父进程
        $this->process->start();
        swoole_process::wait();
    }
    /**
     * @param $worker
     * 任务进程池
     */
    public function run($worker){
        //这里可以执行sql
        $this->current_num = $this ->min_worker_num;
        for($i=0 ;$i< $this->current_num;$i++){
            $process = new swoole_process([$this,'task_num'],false,2);
            $pid = $process->start();
            $this->process_list[$pid] = $process;
            $this->process_use[$pid] = 0;
        }
        foreach($this->process_list as $v){
            swoole_event_add($v->pipe,function($pipe) use ($v){
                $data = $v -> read();
                var_dump($data);
                $this->process_use[$data] = 0;
            });
        }
        /**
         * 每一秒钟去发一次任务
         */
        swoole_timer_tick(1000,function($timer_id){
            static $num = 0;
            $num += 1;
            $flag = true;
            //它去看看哪一个进程是没有被使用的
            foreach($this->process_use as $k => $v ){
                if($v == 0){
                    $flag =false;
                    $this->process_use[$k] = 1; //并且把它标记成1 在给它发个任务
                    $this->process_list[$k] ->write($num ."hello");
                    break;
                }
            }
            //如果所有的worker子进程都再忙着了，再当前进程池还没满的情况下启动一个新的进程池
            if($flag && $this->current_num < $this->max_worker_num){
                $process = new swoole_process([$this,'task_num'],false,2);
                $pid = $process ->start();
                $this->process_list[$pid] = $process ;
                $this->process_use[$pid] = 1;
                $this->process_list[$pid] ->write($num."Hello");
            }
            var_dump($num);//如果执行完十次任务后 关闭当前定时器关闭当前子进程
            if($num == 10){
                foreach($this->process_list as $v){
                    $v -> write('exit');
                }
                swoole_timer_clear($timer_id);
                $this->process->exit();
            }
        });
    }

    //接受主进程的消息
    public function task_num($worker){
        //当读到父进程发送来的任务时
        swoole_event_add($worker->pipe,function($pipe) use ($worker){
            $data = $worker->read();
            //当某个子进程收到任务的时候，它会打印自己的进程号，和它所接到的任务的消息
            var_dump($worker->pid.":".$data);
            if($data =='exit'){
                $worker->exit();exit;
                sleep(5);
                $worker -> write("",$worker->pid);
            }
        });
    }
}
new BaseProcess();
//监听到进程退出了
swoole_process::signal(SIGCHLD,function($sig){
    //必须为false
    while($ret = swoole_process::wait(false)){
    echo "PID = {$ret['pid']}\n";
    }
});